package com.yjs.app.trade.service.service.util;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.UnsupportedEncodingException;
import java.util.HashMap;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.rabbitmq.client.Address;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
import com.yjs.core.mq.message.BaseMessage;

import net.sf.json.JSONObject;

/**
 * 队列控制器
 * 
 * @version 1.0.0
 */
@Component
public class RabbitMQService {

	@Autowired
	UtilService utilService;

	public static Map<String, Channel> map = new HashMap<String, Channel>();

	static Logger log = LoggerFactory.getLogger(Logger.class);
	
	
	
	// 打开连接和创建频道，与发送端一样
	public void routingRecv(String url, int port, String username, String password, String queueName, String durablee,
			String exchangeName, String routingKey) throws Exception {
		boolean durable = Boolean.parseBoolean(durablee);
		ConnectionFactory factory = new ConnectionFactory();
//		factory.setHost(url);
		Address []  addresses;
		if(url.split(",").length == 2){
			String[] urls = url.split(",");
			addresses = new Address[]{
					new Address(urls[0], port),new Address(urls[1], port)
			};
		}else{
			addresses = new Address[]{
					new Address(url, port)
			};
		}
		
		factory.setPort(port);
		factory.setUsername(username);// MQ用户名
		factory.setPassword(password);// MQ密码
		Connection connection = factory.newConnection(addresses);
		Channel channel = connection.createChannel();
		// 声明队列，主要为了防止消息接收者先运行此程序，队列还不存在时创建队列。
		channel.queueDeclare(queueName, durable, false, false, null);
		
		/*Consumer consume=new DefaultConsumer(channel){
			public void handleDelivery(String consumerTag, com.rabbitmq.client.Envelope envelope, 
					com.rabbitmq.client.AMQP.BasicProperties properties, 
					byte[] body) throws IOException {
				
			};
		};*/
		
		// 绑定对列到交换机上,并指定路由键
		channel.queueBind(queueName, exchangeName, routingKey);
		channel.basicQos(1);
		// 创建队列消费者
		QueueingConsumer consumer = new QueueingConsumer(channel);
		// 指定消费队列
		channel.basicConsume(queueName, true, consumer);
		// 指定消费队列
		map.put(queueName, channel);
		log.debug("rabbitMQ已经启动``````" + queueName);
		while (true) {
			// nextDelivery是一个阻塞方法（内部实现其实是阻塞队列的take方法）
			QueueingConsumer.Delivery delivery = consumer.nextDelivery();
			if (channel.isOpen()) {
				try {
					byte[] body = delivery.getBody();
					BaseMessage baseMessage = (BaseMessage) deserialize(body);
					String messageContent = (String) baseMessage.getMessageContent();
					utilService.reviceMsgByRabbit(messageContent);
				} catch (Exception e) {
					e.printStackTrace();
					log.error("消息队列错误:", e);
				}
			}
		}
	}

	public void recv(String url , int port, String username, String password,String queueName) throws Exception {
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost(url);
		factory.setPort(port);
		factory.setUsername(username);//MQ用户名
		factory.setPassword(password);//MQ密码
		Connection connection = factory.newConnection();
		Channel channel = connection.createChannel();
		//声明队列，主要为了防止消息接收者先运行此程序，队列还不存在时创建队列。
		channel.queueDeclare(queueName, false, false, false, null);
		//创建队列消费者
		QueueingConsumer consumer = new QueueingConsumer(channel);
		//指定消费队列
		channel.basicConsume(queueName, true, consumer);
		//指定消费队列
		map.put(queueName, channel);
		log.debug("rabbitMQ已经启动``````" + queueName);
		while (true) {
			//nextDelivery是一个阻塞方法（内部实现其实是阻塞队列的take方法）
			QueueingConsumer.Delivery delivery = consumer.nextDelivery();
			if (channel.isOpen()) {
				try {
					String message = new String(delivery.getBody());
					JSONObject jsonObject = JSONObject.fromObject(message);
					log.debug(jsonObject.toString());
				} catch (Exception e) {
					e.printStackTrace();
					log.error("消息队列错误:",e);
				}
			}
		}
	}
	
	public static Object deserialize(byte[] b) {
		try {
			ObjectInputStream oin = new ObjectInputStream(new ByteArrayInputStream(b));
			try {
				return oin.readObject();
			} catch (ClassNotFoundException var3) {
				// logger.error("deserialize the object
				// class[ClassNotFoundException]error");
				return null;
			}
		} catch (IOException var4) {
			var4.printStackTrace();
			return null;
		}

	}

	public static void send(String message, String code) throws Exception {
		if (map != null && map.size() > 0 && map.get(code) != null && map.get(code).isOpen()) {
			// 指定一个队列
			map.get(code).queueDeclare(code, true, false, false, null);
			// 往队列中发出一条消息
			map.get(code).basicPublish("", code, null, message.getBytes());
		}
	}

	public static void main(String[] args) throws UnsupportedEncodingException {
		byte[] b = { 123, 34, 109, 115, 103, 84, 121, 112, 101, 34, 58, 49, 44, 34, 105, 115, 99, 104, 101, 99, 107, 34, 58, 49, 44, 34, 109, 101, 115, 115, 97, 103, 101, 67, 111, 110, 116, 101, 110, 116, 34, 58, 34, 97, 100, 97, 115, 100, 97, 115, 100, 34, 44, 34, 109, 115, 103, 78, 117, 109, 98, 101, 114, 34, 58, 34, 50, 50, 50, 48, 49, 56, 48, 54, 49, 50, 51, 54, 48, 51, 56, 95, 48, 48, 57, 34, 44, 34, 115, 116, 97, 116, 117, 115, 34, 58, 49, 44, 34, 109, 101, 109, 98, 101, 114, 84, 121, 112, 101, 34, 58, 49, 125 };
		BaseMessage baseMessage = (BaseMessage) deserialize(b);
		/* String message = new String(b); */
		System.out.println(baseMessage.getMessageContent());
	}
}
